package cars;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Set;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Manlikecar {

	public static void main(String[] args) throws Exception{
		if (args.length!=3) {
			System.err.println("path err!!!");
			System.exit(-1);
		}
		
		Job job=new Job(new Configuration(), "manlike");
		job.setJarByClass(Manlikecar.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		job.setMapperClass(mamap.class);
		job.setReducerClass(mareduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		job.waitForCompletion(true);
		
		Job job2=new Job();
		job2.setJarByClass(Manlikecar.class);
		FileInputFormat.addInputPath(job2, new Path(args[1]));//将第一组的输出路径作为输入路径
		FileOutputFormat.setOutputPath(job2, new Path(args[2]));
		job2.setMapperClass(sortMap.class);
		job2.setReducerClass(sortReduce.class);
		job2.setMapOutputKeyClass(IntWritable.class);
		job2.setMapOutputValueClass(Text.class);
		job2.setOutputKeyClass(Text.class);
		job2.setOutputValueClass(Text.class);
		job2.waitForCompletion(true);

	}
	//the first group
	public static class mamap extends Mapper<LongWritable, Text, Text,IntWritable>{
		protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,IntWritable>.Context context) throws java.io.IOException ,InterruptedException {
			String[] lines = value.toString().split("\t");
			if (lines.length>38 && lines[38]!="" && lines[38]!=null) {
				if(lines[38].trim().equals("男性")){//判断性别，获取男性买车的品牌
					if(lines[7]!=null){
						String band=lines[7].trim();
						context.write(new Text(band), new IntWritable(1));
					}
				}
			}
			
		};
	}
	//map产生的中间结果是：《东风，1》
	
	//map产生的结果经过shuffle处理，处理后的格式为：《东风，『1，1，1，1，1』》
	public static class mareduce extends Reducer<Text, IntWritable, Text, IntWritable>{
		protected void reduce(Text k2, java.lang.Iterable<IntWritable> values, org.apache.hadoop.mapreduce.Reducer<Text,IntWritable,Text,IntWritable>.Context context) throws java.io.IOException ,InterruptedException {
			int sum=0;
			for (IntWritable count : values) {
				sum+=count.get();
			}
			context.write(k2,new IntWritable(sum));
		};
	}
	//reduce产生的最后结果：《东莞，100》.
	//reduce产生的最后结果要存放在hdfs，其中，数据的分割符是\t.
	
	//the second group
	//第二组的map函数的输入应该是第一组reduce的输出。
	public static class sortMap extends Mapper<LongWritable, Text,IntWritable,Text>{
		protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,IntWritable,Text>.Context context) throws java.io.IOException ,InterruptedException {
			String[] lines = value.toString().split("\t");
				String band=lines[0];
				int amount=Integer.parseInt(lines[1].trim());
				context.write(new IntWritable(amount),new Text(band));
		};
	}
	//map的中间结果经过shuffle阶段处理，处理之后变成以下格式：<100,{东风....}>
	
	public static class sortReduce extends Reducer<IntWritable,Text,Text,Text>{
		private static TreeMap<Integer,String> clMap=new TreeMap<Integer,String>(
				 new Comparator<Integer>() {

					@Override
					public int compare(Integer o1, Integer o2) {
						// TODO Auto-generated method stub
						return o2.compareTo(o1);
					}
				});
		protected void reduce(IntWritable k3, java.lang.Iterable<Text> values, org.apache.hadoop.mapreduce.Reducer<IntWritable,Text,Text,Text>.Context context) throws java.io.IOException ,InterruptedException {
			for (Text brand : values) {
				clMap.put(k3.get(), brand.toString());
			}
			if (clMap.size()>5) {
				clMap.remove(clMap.lastKey());
			}
		};
		protected void cleanup(org.apache.hadoop.mapreduce.Reducer<IntWritable,Text,Text,Text>.Context context) throws java.io.IOException ,InterruptedException {
			ArrayList<String> list=new ArrayList<String>();
			Set<Integer> keySet = clMap.keySet();//获取treemap集合的所有key
			for (Integer key : keySet) {
				list.add(clMap.get(key)+"--"+key);//将最受欢迎的品牌输入到集合中
			}
			context.write(new Text("2013年最受男性欢迎的前5个汽车品牌"), new Text(list.toString()));
		};
		
	
	}

}
